package com.hadoop.sxtmr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class RunJob {
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		job.setJarByClass(RunJob.class);

		// 指定用户自定义的Mapper组件
		job.setMapperClass(HotMapper.class);

		// 指定用户自定义的Reducer组件
		job.setReducerClass(HotReducer.class);

		// 设置Mapper输出key类型
		job.setMapOutputKeyClass(KeyPair.class);
		// 设置Mapper输出value类型，注意导包是Hadoop包
		job.setMapOutputValueClass(Text.class);

		// 设置Reducer输出的key类型
		job.setOutputKeyClass(KeyPair.class);
		// 设置Reducer输出的value类型
		job.setOutputValueClass(Text.class);

		// 指定处理文件的HDFS读取路径
		FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.186.6:9000/mr2/hot.txt"));
		// 指定输出结果文件的所在目录，注意，指定的此目录事先不能存在
		FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.186.6:9000/mr2/result/hot/"));
		
		job.setNumReduceTasks(3);
		
		//分区
		job.setPartitionerClass(Firstpartition.class);
		//使用自定义排序
		job.setSortComparatorClass(SortHot.class);
		//使用自定义分组
		job.setGroupingComparatorClass(GroupHot.class);
		
		job.waitForCompletion(true);
	}
}
